package club.hicode.daydayup.zookeeper.distribute4;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import com.github.sd4324530.jtuple.Tuple2;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.math.IEEE754rUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.springframework.amqp.core.FanoutExchange;

import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;

/**
 * 原生 Zookeeper 分布式锁
 *
 * @author Liu Chunfu
 * @date 2018-06-11 上午10:50
 **/
@Slf4j
@Data
public class ZkDistrubuteLock implements Watcher {

    private static final String CONNECT_STR = "39.108.231.92:2181";

    private static final String LOCK_ROOT = "/lock";

    private CountDownLatch startCdl = new CountDownLatch(1);

    private CountDownLatch workCdl = new CountDownLatch(1);

    private String currentPath;

    private String preListenPath;

    private ZooKeeper zooKeeper;


    public ZkDistrubuteLock() throws Exception {
        zooKeeper = new ZooKeeper(CONNECT_STR, 5 * 10 * 1000, this::process);
        startCdl.await();
        Stat exists = zooKeeper.exists(LOCK_ROOT, false);
        if (exists == null) {
            zooKeeper.create(LOCK_ROOT, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }


    public void lock() {
        try {
            //创建一个节点,注意路径需要以 / 结尾
            currentPath = createNode();

            Tuple2<Boolean, String> tuple2 = predicateMinCode(currentPath);
            if (tuple2.first) {
                log.debug("获取锁成功..");
                return;
            }
            preListenPath = tryWrapper(tuple2.second);
            zooKeeper.exists(preListenPath, true);
            log.debug("在{}中监听路径{}", currentPath, preListenPath);

            workCdl.await();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public void unlock() {
        try {
            log.debug("删除节点:{}", currentPath);
            zooKeeper.delete(currentPath, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    /**
     * 事件触发
     *
     * @param event
     */
    @Override
    public void process(WatchedEvent event) {
        Event.KeeperState state = event.getState();

        Event.EventType type = event.getType();

        if (state == Event.KeeperState.SyncConnected) {
            startCdl.countDown();
        }

        if (type == Event.EventType.NodeDeleted) {
            workCdl.countDown();
        }
    }

    /**
     * 创建节点
     *
     * @throws KeeperException
     * @throws InterruptedException
     */
    private String createNode() throws KeeperException, InterruptedException {
        return zooKeeper.create(LOCK_ROOT + "/", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    }

    /**
     * 在路径前面加上 /root/
     *
     * @param path
     * @return
     */
    private String tryWrapper(String path) {
        String parentPath = LOCK_ROOT + "/";
        //
        if (!path.startsWith(parentPath)) {
            return parentPath + path;
        }
        return path;
    }

    /**
     * 去掉路径前面部分parent 修饰符
     *
     * @param path
     * @return
     */
    private String unwrapper(String path) {
        return StrUtil.removePrefix(path, LOCK_ROOT + "/");
    }

    /**
     * 预测指定节点是否是最小节点，如果不是返回最小节点
     *
     * @param path 需要判定的路径
     * @return
     */
    private Tuple2<Boolean, String> predicateMinCode(String path) {
        path = tryWrapper(path);
        try {
            List<String> children = zooKeeper.getChildren(LOCK_ROOT, false);
            TreeSet<String> treeSet = new TreeSet<>(children);
            String smallest = LOCK_ROOT + "/" + treeSet.first();
            if (smallest.equals(path)) {
                return Tuple2.with(true, null);
            }
            //进行预测，注意获取的是排序后的最后一个。
            SortedSet<String> headSet = treeSet.headSet(unwrapper(path));
            if (CollectionUtil.isEmpty(headSet)) {
                log.error("异常情况。。。");
                return null;
            }
            return Tuple2.with(false, headSet.last());
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }
}
